package cn.doitedu.dashboard;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job03_DashBoard_GMV_01 {

    public static void main(String[] args) {

        // 创建编程环境

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);


        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        TableConfig config = tenv.getConfig();
        config.set("table.exec.source.idle-timeout","1 ms");

        // 开启窗口的提前触发器
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled", true);
        // 把触发器的触发周期设置为1秒
        config.getConfiguration().setString("table.exec.emit.early-fire.delay", "1s");


        // - 今天累计到当前的gmv总额，应付(或实付)总额，优惠总额（要求每秒更新一次）
        // 建表，映射 业务库中的 订单表

        tenv.executeSql(
                " create table oms_order_cdc(         "+
                        "    id bigint,                            "+  // 订单id
                        "    total_amount decimal(10,2),           "+  // 订单总金额
                        "    pay_amount decimal(10,2),           "+  // 订单应付(实付)金额
                        "    promotion_amount decimal(10,2),       "+  // 促销减免金额
                        "    integration_amount decimal(10,2),     "+  // 积分抵扣金额
                        "    coupon_amount decimal(10,2),          "+  // 优惠券抵扣金额
                        "    discount_amount decimal(10,2),        "+  // 后台折扣金额
                        "    create_time timestamp(3),             "+  // 订单创建时间
                        "    modify_time timestamp(3),             "+  // 数据更新时间

                        "    rt as modify_time,                   "+
                        "    watermark for rt as rt ,             "+

                        "    primary key(id) not enforced     "+
                        " ) WITH (                            "+
                        "     'connector' = 'mysql-cdc',      "+
                        "     'hostname' = 'doitedu',         "+
                        "     'port' = '3306',                "+
                        "     'username' = 'root',            "+
                        "     'password' = 'root',            "+
                        "     'database-name' = 'realtimedw', "+
                        "     'table-name' = 'oms_order'      "+
                        " )                                   "
        );


        // 建表，映射  看板库中的 目标表
//        tenv.executeSql("select * from oms_order_cdc").print();
//        System.exit(1);
//


        // 计算 输出
        // StreamPhysicalWindowAggregate doesn't support consuming update and delete changes which is produced by node TableSourceScan
        tenv.executeSql(
                        " with tmp as (                                                                                      "+
                        "      select  * from oms_order_cdc                                                                  "+
                        "      where   date_format(create_time,'yyyy-MM-dd') = current_date                                  "+
                        "  )                                                                                                 "+
                        " SELECT                                                                                             "+
                        "      tumble_start(rt,interval '24' hour)                                     "+
                        "     ,tumble_end(rt,interval '24' hour)                                       "+
                        "     ,sum(total_amount) as total_amount                                                             "+
                        "     ,sum(pay_amount) as pay_amount                                                                 "+
                        "     ,sum(promotion_amount+integration_amount+coupon_amount+discount_amount) as discount_amount     "+
                        "     ,CURRENT_TIMESTAMP as calc_time     "+
                        " FROM tmp                                                                                           "+
                        " group by                                                                                           "+
                        "      tumble( rt, interval '24' hour)                                                                 "
              ).print();

    }

}
